#include "ndk/epoll_reactor.h"

#if defined NDK_LINUX
namespace ndk
{
int epoll_reactor_handler_repository::open(int size)
{
  if (size <= 0) return -1;
  this->handlers_   = new epoll_event_tuple[size];
  if (this->handlers_ == 0)
    return -1;
  this->max_size_ = size;
  // Try to set the number of handles to the current limit. 
  // Deny user to spawn new handle that overtop of the size, it'll overflow
  return 0;//ndk::set_handle_limit(this->max_size_);
}
int epoll_reactor_handler_repository::close()
{
  if (this->handlers_ != 0)
  {
    this->unbind_all();
    delete []this->handlers_;
    this->handlers_  = 0; 
    this->max_size_  = 0;
  }
  return 0;    
}
int epoll_reactor_handler_repository::unbind_all()
{
  for (int handle = 0; handle < this->max_size_; ++handle)
    this->unbind(handle);
  return 0;
}
//---------------------------------------------------------------------------
epoll_reactor_notify::epoll_reactor_notify()
: ep_reactor_(0),
  notification_pipe_()
{
}
int epoll_reactor_notify::open(reactor_impl *impl)
{
  STRACE("epoll_reactor_notify");
  this->ep_reactor_ = dynamic_cast<epoll_reactor *>(impl);
  if (this->ep_reactor_ == 0) return -1;
  if (this->notification_pipe_.open() == -1) return -1;
  // 
  // close-on-exec
  ::fcntl(this->notification_pipe_.read_handle(), F_SETFD, FD_CLOEXEC);
  ::fcntl(this->notification_pipe_.write_handle(), F_SETFD, FD_CLOEXEC);

  ndk::set_non_block_mode(this->notification_pipe_.read_handle());
  ndk::set_non_block_mode(this->notification_pipe_.write_handle());
  return 0;
}
int epoll_reactor_notify::close()
{
  STRACE("epoll_reactor_notify");
  return this->notification_pipe_.close();
}
int epoll_reactor_notify::notify(event_handler *eh/* = 0*/,
                                 reactor_mask mask/* = event_handler::except_mask*/)
{
  STRACE("epoll_reactor_notify");
  if (this->ep_reactor_ == 0) return -1;

  // 
  notification_buffer buff(eh, mask);
  return ndk::write(this->notification_pipe_.write_handle(), 
                    (void *)&buff, 
                    sizeof(buff)) > 0 ? 0 : -1;
}
int epoll_reactor_notify::purge_pending_notifications(event_handler * /*eh = 0*/,
                                                      reactor_mask/* = event_handler::all_events_mask*/)
{
  STRACE("epoll_reactor_notify");
  return -1;
}
// ------------------------------------------------------------------------
epoll_reactor::epoll_reactor()
: initialized_(false),
  epoll_fd_(NDK_INVALID_HANDLE),
  size_(0),
  curr_payload_(0),	   
  deactivated_(0),
  delete_timer_queue_(0),
  events_(0),
  start_pevents_(0),
  end_pevents_(0),
  timer_queue_(),
  notify_handler_(0),
  owner_(NULL_thread),
  token_(this)
{
}
epoll_reactor::~epoll_reactor()
{
  STRACE("epoll_reactor");
  this->close();
}
int epoll_reactor::open(int size/* = 0*/, timer_queue *tq/* = 0*/, 
                        bool reset_fd_limit/* = true*/)
{
  STRACE("epoll_reactor");
  guard<epoll_reactor_token> g(this->token_);
  if (this->initialized_)
    return -1;

  this->timer_queue_ = tq;
  this->size_        = size;
  if (this->size_ == 0)
    this->size_ = ndk::max_handles();
  else if (reset_fd_limit)
  {
    ndk::set_handle_limit(this->size_);
  }

  if (this->size_ <= 0) return -1;

  int result = 0;
  this->notify_handler_ = new epoll_reactor_notify();
  if (this->notify_handler_ == 0)
  {
    this->close(); return -1;
  }
  this->events_ = new epoll_event[this->size_];
  if (this->events_ == 0)
  {
    this->close(); return -1;
  }

  // Since Linux 2.6.8, the size argument is unused.
  this->epoll_fd_ = ::epoll_create(this->size_);
  if (this->epoll_fd_ == NDK_INVALID_HANDLE)
  {
    this->close(); return -1;
  }
  ::fcntl(this->epoll_fd_, F_SETFD, FD_CLOEXEC);

  if (this->handler_rep_.open(ndk::max_handles()))
  {
    this->close(); return -1;
  }
  if (this->timer_queue_ == 0)
  {
    this->timer_queue_ = new timer_queue();
    if (this->timer_queue_ == 0)
    {
      this->close(); return -1;
    }else this->delete_timer_queue_ = 1;
  }

  if (this->notify_handler_->open(this) == -1
      || 
      this->register_handler_i(this->notify_handler_->notify_handle(),
                               this->notify_handler_,
                               event_handler::read_mask) == -1)
    result = -1;

  if (result != -1) 
    this->initialized_ = true;
  else 
    this->close();

  return result;
}
int epoll_reactor::close(void) 
{
  STRACE("epoll_reactor");
  guard<epoll_reactor_token> g(this->token_);
  if (this->epoll_fd_ != NDK_INVALID_HANDLE)
    ndk_os::close(this->epoll_fd_);
  this->epoll_fd_ = NDK_INVALID_HANDLE;
  if (this->events_)
    delete [] this->events_;
  this->events_ = 0;
  if (this->timer_queue_)
    delete this->timer_queue_;
  this->timer_queue_ = 0;
  if (this->notify_handler_)
    delete this->notify_handler_;
  this->notify_handler_ = 0;

  this->events_ = 0;
  this->handler_rep_.close();
  this->epoll_fd_      = NDK_INVALID_HANDLE;
  this->curr_payload_  = 0;
  this->start_pevents_ = 0;
  this->end_pevents_   = 0;
  this->initialized_   = false;
  this->deactivated_   = 1;
  return 0;
}
int epoll_reactor::work_pending_i(const time_value *max_wait_time)
{
  STRACE("epoll_reactor");
  if (this->deactivated_) return -1;
  if (this->start_pevents_ != this->end_pevents_)
    return 1;  // We still have work_pending(). Do not poll for
  // additional events.

  time_value wait_time(0, 0);
  time_value *this_timeout = 0;
  this_timeout = this->timer_queue_->calculate_timeout(max_wait_time, 
                                                       &wait_time);

  // Check if we have timers to fire.
  const int timers_pending =
    ((this_timeout != 0 && max_wait_time == 0)  // one timer
     || (this_timeout != 0 && max_wait_time != 0
         && *this_timeout != *max_wait_time) ? 1 : 0); 

  const int timeout =
    (this_timeout == 0 ? -1 /* Infinity */
     : static_cast<int>(this_timeout->msec())/* Point to wait_time */);	

  // Wait for events
  int nfds = 0;
  nfds = ::epoll_wait(this->epoll_fd_,
                      this->events_,
                      this->size_,
                      static_cast<int>(timeout));
  if (nfds > 0)
  {
    this->start_pevents_ = this->events_;
    this->end_pevents_   = this->start_pevents_ + nfds;
  }
  // If timers are pending, override any timeout from the epoll.
  return((nfds == 0 && timers_pending != 0) ? 1 : nfds);
}
int epoll_reactor::handle_events(const time_value *max_wait_time/* = 0*/) 
{
  STRACE("epoll_reactor");
  token_guard g(this->token_);
  int result = g.acquire_quietly(max_wait_time);
  if (!g.is_owner())
    return result;

  if (this->deactivated_) 
    return -1;

  return this->handle_events_i(max_wait_time);
}
int epoll_reactor::handle_events_i(const time_value *max_wait_time) 
{
  STRACE("epoll_reactor");
  int result  = 0;
#if 0
  int err_num = 0;
  time_value max_wait_time_buff;
  if (max_wait_time)
    max_wait_time_buff = *max_wait_time;

  /**
   * Wait pending events use epoll_wait, but not check errno
   */
  time_value tp_before, tp_after;
  do
  {
    if (max_wait_time != 0)
      tp_before.gettimeofday();
    //
    result = this->work_pending_i(max_wait_time ? &max_wait_time_buff : 0);
    //
    err_num = ndk_os::last_error();
    // This code block used to save epoll_wait escape time
    if (result == -1 && err_num == EINTR)
    {
      if (max_wait_time != 0)
      {
        if (*max_wait_time > time_value::zero)
        {
          tp_after.gettimeofday();
          max_wait_time_buff -= (tp_after - tp_before);
          if (max_wait_time_buff < time_value::zero)
          {
            result = 0;  // timeout
            break;
          }
        }else
          break;
      }
    }
  }while (result == -1 && err_num == EINTR);
#else
  do
  {
    result = this->work_pending_i(max_wait_time);
  }while (result == -1 && errno == EINTR);
#endif

  // timeout
  if (result == 0)
    return 0;
  else if (result == -1 && errno != EINTR)
    return result;
  // Dispatch an event.
  return this->dispatch();
}
int epoll_reactor::dispatch()
{
  STRACE("epoll_reactor");
  int result = 0;
  // Handle timers early since they may have higher latency
  // constraints than I/O handlers.  Ideally, the order of
  // dispatching should be a strategy...

  result += this->dispatch_timer_handler();

  // Check to see if there are no more I/O handles left to
  // dispatch AFTER we've handled the timers.
  //
  result += this->dispatch_io_event();
  return result;
}
int epoll_reactor::dispatch_timer_handler()
{
  STRACE("epoll_reactor");
  return this->timer_queue_->expire(ndk::gettimeofday());
}
int epoll_reactor::dispatch_io_event()
{
  STRACE("epoll_reactor");
  struct epoll_event *& pfd = this->start_pevents_;
  int result = 0;
  while (pfd < this->end_pevents_ && result < MAX_REACTOR_PROCESS_FDS_ONE_TIME)
  {
    // 
    event_handler *eh = this->handler_rep_.find(pfd->data.fd);
    if (eh)
    {
      // Note that if there's an error(such as the handle was closed
      // without being removed from the event set) the EPOLLHUP and/or
      // EPOLLERR bits will be set in pfd->events.
      if (NDK_BIT_ENABLED(pfd->events, EPOLLIN))
      {
        NDK_CLR_BITS(pfd->events, EPOLLIN);
        if (this->upcall(eh, 
                         &event_handler::handle_input, 
                         pfd->data.fd) < 0)
          this->remove_handler_i(pfd->data.fd, 
                                 event_handler::read_mask);
        ++result;
      }else if (NDK_BIT_ENABLED(pfd->events, EPOLLOUT))
      {
        NDK_CLR_BITS(pfd->events, EPOLLOUT);
        if (this->upcall(eh, 
                         &event_handler::handle_output, 
                         pfd->data.fd) < 0)
          this->remove_handler_i(pfd->data.fd, 
                                 event_handler::write_mask);
        ++result;
      }else if (NDK_BIT_ENABLED(pfd->events, EPOLLPRI))
      {
        NDK_CLR_BITS(pfd->events, EPOLLPRI);
        if (this->upcall(eh, 
                         &event_handler::handle_exception, 
                         pfd->data.fd) < 0)
          this->remove_handler_i(pfd->data.fd, 
                                 event_handler::except_mask);
        ++result;
      }else if (NDK_BIT_ENABLED(pfd->events, EPOLLHUP | EPOLLERR))
      {
        this->remove_handler_i(pfd->data.fd, 
                               event_handler::all_events_mask);
        ++result;
      }else
      {
        NDK_LOG("dispatch_io [handle = %d] trigger unknown events 0x%x",
                pfd->data.fd, pfd->events);
        ++pfd;
      } 
      // If more than one event comes in between epoll_wait(2) calls,
      // they will be combined reported.
      if (pfd->events == 0)
        ++pfd; 
    }else // if (eh)
    {
      NDK_LOG("dispatch_io [handle = %d] not match event handler", 
              pfd->data.fd);
      ++pfd; 
    }
  } // while (pfd < this->end_pevents_ ...
  return result;
}
int epoll_reactor::register_handler_i(ndk_handle handle,
                                      event_handler *event_handler,
                                      reactor_mask mask)
{
  STRACE("epoll_reactor");
  if (handle == NDK_INVALID_HANDLE
      || event_handler == 0
      || mask == event_handler::null_mask)
    return -1;
  if (this->handler_rep_.find(handle) == 0)
  {
    // Handler not present in the repository.  Bind it.
    if (this->handler_rep_.bind(handle, event_handler, mask) != 0)
      return -1;
    struct epoll_event epev;
    std::memset(&epev, 0, sizeof(epev));
    epev.events  = this->reactor_mask_to_epoll_event(mask);
    epev.data.fd = handle;
    if (::epoll_ctl(this->epoll_fd_, EPOLL_CTL_ADD, handle, &epev) == -1)
    {
      this->handler_rep_.unbind(handle);
      return -1;
    }
    this->incre_payload();
  }else
  {
    // Handler is already present in the repository, so register it
    // again, possibly for different event.  Add new mask to the
    // current one
    return this->mask_opt_i(handle, mask, epoll_reactor::add_mask);
  }
  return 0;
}
size_t epoll_reactor::reactor_mask_to_epoll_event(reactor_mask mask)
{
  STRACE("epoll_reactor");
  size_t events = 0;
  if (NDK_BIT_ENABLED(mask, event_handler::read_mask))
    NDK_SET_BITS(events, EPOLLIN);

  if (NDK_BIT_ENABLED(mask, event_handler::epoll_edge))
    NDK_SET_BITS(events, EPOLLET);

  if (NDK_BIT_ENABLED(mask, event_handler::accept_mask))
    NDK_SET_BITS(events, EPOLLIN);

  if (NDK_BIT_ENABLED(mask, event_handler::connect_mask))
    NDK_SET_BITS(events, EPOLLIN|EPOLLOUT);

  if (NDK_BIT_ENABLED(mask, event_handler::write_mask))
    NDK_SET_BITS(events, EPOLLOUT);
  // EXCEPT flag will place the handle in the except set.
  if (NDK_BIT_ENABLED(mask, event_handler::except_mask))
    NDK_SET_BITS(events, EPOLLPRI);

  return events;
}
int epoll_reactor::mask_opt_i(ndk_handle handle,
                              reactor_mask mask,
                              int opt)
{
  STRACE("epoll_reactor");
  reactor_mask new_mask = this->handler_rep_.mask(handle);
  switch(opt)
  {
  case epoll_reactor::add_mask: NDK_SET_BITS(new_mask, mask); break;
  case epoll_reactor::set_mask: new_mask = mask; break;
  case epoll_reactor::clr_mask: NDK_CLR_BITS(new_mask, mask); break;
  default: return -1;
  }
  this->handler_rep_.mask(handle, new_mask);

  struct epoll_event epev;
  std::memset(&epev, 0, sizeof(epev));
  epev.data.fd = handle;

  if (new_mask == event_handler::null_mask)
  {
    epev.events = 0;
    return ::epoll_ctl(this->epoll_fd_, EPOLL_CTL_DEL, handle, &epev);	
  }else
  {
    epev.events = this->reactor_mask_to_epoll_event(new_mask);
    if (::epoll_ctl(this->epoll_fd_, EPOLL_CTL_MOD, handle, &epev) == -1)
    {
      // If a handle is closed, epoll removes it from the poll set
      // automatically - we may not know about it yet. If that's the
      // case, a mod operation will fail with ENOENT. Retry it as
      // an add.
      if (ndk_os::last_error() == ENOENT)
        return ::epoll_ctl(this->epoll_fd_, EPOLL_CTL_ADD, handle, &epev);
    }
  }
  return 0;
}
int epoll_reactor::remove_handler_i(ndk_handle handle,
                                    reactor_mask mask)
{
  STRACE("epoll_reactor");
  event_handler *eh = this->handler_rep_.find(handle);
  if (eh == 0 ||
      this->mask_opt_i(handle, mask, epoll_reactor::clr_mask) == -1)
    return -1;
  if (NDK_BIT_DISABLED(mask, event_handler::dont_call))
    eh->handle_close(handle, mask);
  // If there are no longer any outstanding events on the given handle
  // then remove it from the handler repository.
  if (this->handler_rep_.mask(handle) == event_handler::null_mask)
  {
    this->handler_rep_.unbind(handle);
    {
      this->decre_payload();
    }
  }
  return 0;
}
}// namespace ndk
#endif // NDK_LINUX

